package com.example.demo.zaqizabatest.completablefuture;

        import java.util.ArrayList;
        import java.util.List;
        import java.util.concurrent.*;
        import java.util.function.Supplier;

/**
 * @ClassName CompletableFutureTest
 * @Author lys4134
 * @Date 2021/12/24 09:47:20
 * @Description CompletableFutureTest
 * @Version 1.0
 **/
public class CompletableFutureTest {

    private static final ExecutorService executorService = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors() < 1 ? 1 : Runtime.getRuntime().availableProcessors());

    public static void main(String[] args) throws Exception {
//        long start = System.currentTimeMillis();
//        CompletableFuture<String> completableFutureBySupplier = testCreateCompletableFutureBySupplier(() -> "-------------创建Completable----------", null);
//        testThenApply(completableFutureBySupplier, "-------------测试thenApply----------");
//        testThenRun(completableFutureBySupplier, "-------------测试thenRun----------");
//        testThenCombine(completableFutureBySupplier, "-------------测试thenCombineAsync----------");
//        System.out.println("Main函数耗时：" + (System.currentTimeMillis() - start));
//        testWhenComplete();
//        testThenCompose();
//        testAllOf();
        testAllOfAndThreadPoll();
    }

    private static void testAllOfAndThreadPoll() {
        Long start = System.currentTimeMillis();
        ArrayList<Future> futures = new ArrayList<>();
        for (int i = 0; i < 3; i++) {
            futures.add(CompletableFuture.runAsync(() -> {
                System.out.println("testAllOfAndThreadPoll: " + Thread.currentThread().getName());
                try {
                    Thread.sleep(3000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                try {
                    testThreadPoll();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }));
        }
        // 异步执行
        CompletableFuture.allOf((CompletableFuture<?>[]) futures.toArray(new CompletableFuture<?>[0])).join();
        System.out.println(System.currentTimeMillis() - start);
    }

    private static void testThreadPoll() throws InterruptedException {
        ExecutorService fixedThreadPool = Executors.newFixedThreadPool(3);
        CountDownLatch latch = new CountDownLatch(3);
        for (int i = 0; i < 3; i++) {
            fixedThreadPool.execute(() -> {
                System.out.println("testThreadPoll: " + Thread.currentThread().getName());
                try {
                    Thread.sleep(5000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                latch.countDown();
            });
        }
        latch.await();
        fixedThreadPool.shutdown();
    }

    /**
     *
     * 这两方法各有一个重载版本，可以指定执行异步任务的Executor实现，
     * 如果不指定，默认使用ForkJoinPool.commonPool()，如果机器是单核的，
     * 则默认使用ThreadPerTaskExecutor，该类是一个内部类，每次执行execute都会创建一个新线程。
     *
     * @param runnable
     * @param executor
     * @return
     */
    private static CompletableFuture<Void> testCreateCompletableFutureByRunnable(Runnable runnable, Executor executor) {
        if(executor == null) {
            return CompletableFuture.runAsync(runnable);
        } else {
            return CompletableFuture.runAsync(runnable, executor);
        }
    }

    /**
     *
     * 这两方法各有一个重载版本，可以指定执行异步任务的Executor实现，
     * 如果不指定，默认使用ForkJoinPool.commonPool()，如果机器是单核的，
     * 则默认使用ThreadPerTaskExecutor，该类是一个内部类，每次执行execute都会创建一个新线程。
     *
     * @param supplier
     * @param executor
     * @return
     */
    private static <T> CompletableFuture<T> testCreateCompletableFutureBySupplier(Supplier<T> supplier, Executor executor) throws ExecutionException, InterruptedException {
        long start = System.currentTimeMillis();
        if(executor == null) {
            CompletableFuture<T> future = CompletableFuture.supplyAsync(supplier);
            System.out.println(future.get() + "耗时" + (System.currentTimeMillis() - start));
            return future;
        } else {
            CompletableFuture<T> future = CompletableFuture.supplyAsync(supplier, executor);
            System.out.println(future.get() + "耗时" + (System.currentTimeMillis() - start));
            return future;
        }
    }

    private static<T> CompletableFuture<T> testThenApply(CompletableFuture<T> completableFuture, T t) throws ExecutionException, InterruptedException {
        long start = System.currentTimeMillis();
        CompletableFuture<T> thenApply = completableFuture.thenApply((i) -> t);
        System.out.println(thenApply.get() + "耗时" + (System.currentTimeMillis() - start));
        return thenApply;
    }

    private static<T> CompletableFuture<Void> testThenRun(CompletableFuture<T> completableFuture, T t) throws ExecutionException, InterruptedException {
        long start = System.currentTimeMillis();
        CompletableFuture<Void> thenApply = completableFuture.thenRun(() -> System.out.println(t + "耗时" + (System.currentTimeMillis() - start)));
//        System.out.println(thenApply.get() + "耗时" + (System.currentTimeMillis() - start));
        return thenApply;
    }

    private static<T> CompletableFuture<Object> testThenCombine(CompletableFuture<T> completableFuture, T t) throws ExecutionException, InterruptedException {
        long start = System.currentTimeMillis();
        CompletableFuture<Object> thenApply = completableFuture.thenCombine(CompletableFuture.completedFuture(t), (i,j) -> {
            System.out.println(i + "\n" + j + "耗时" + (System.currentTimeMillis() - start));
            return i + "\n" + j;
        });
//        System.out.println(thenApply.get() + "耗时" + (System.currentTimeMillis() - start));
        return thenApply;
    }

    private static void testWhenComplete() throws ExecutionException, InterruptedException {
        long start = System.currentTimeMillis();
        List<CompletableFuture<String>> futures = new ArrayList<>();
        for (int i =1 ;i <=10;i++){
            final String j = "---->" + i;
            CompletableFuture<String> completableFuture = testCreateCompletableFutureBySupplier(() -> {
                try {
                    Thread.sleep(10);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                String name = Thread.currentThread().getName() + j;
                return name;
            }, executorService);
            futures.add(completableFuture);
        }
        CompletableFuture<Void> completableFuture = CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()]));/*.whenComplete((r, e) -> {
            if (e != null) {
                throw new RuntimeException();
            }
        })*/
        System.out.println(completableFuture.get());
        for (CompletableFuture future : futures) {
            System.out.println(future.get());
        }
//        executorService.shutdown();
        System.out.println("耗时" + (System.currentTimeMillis() - start));
    }


    private static void testWhenComplete1() throws ExecutionException, InterruptedException {
        // 创建异步执行任务:
        CompletableFuture<Double> cf = CompletableFuture.supplyAsync(()->{
            System.out.println(Thread.currentThread()+"job1 start,time->"+System.currentTimeMillis());
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
            }
            if(false){
                throw new RuntimeException("test");
            }else{
                System.out.println(Thread.currentThread()+"job1 exit,time->"+System.currentTimeMillis());
                return 1.2;
            }
        });
        //cf执行完成后会将执行结果和执行过程中抛出的异常传入回调方法，如果是正常执行的则传入的异常为null
        CompletableFuture<Double> cf2=cf.whenComplete((a,b)->{
            System.out.println(Thread.currentThread()+"job2 start,time->"+System.currentTimeMillis());
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
            }
            if(b!=null){
                System.out.println("error stack trace->");
                b.printStackTrace();
            }else{
                System.out.println("run succ,result->"+a);
            }
            System.out.println(Thread.currentThread()+"job2 exit,time->"+System.currentTimeMillis());
        });
        //等待子任务执行完成
        System.out.println("main thread start wait,time->"+System.currentTimeMillis());
        //如果cf是正常执行的，cf2.get的结果就是cf执行的结果
        //如果cf是执行异常，则cf2.get会抛出异常
        System.out.println("run result->"+cf2.get());
        System.out.println("main thread exit,time->"+System.currentTimeMillis());
    }

    /**
     *
     * thenCombine / thenAcceptBoth / runAfterBoth
     * 这三个方法都是将两个CompletableFuture组合起来，只有这两个都正常执行完了才会执行某个任务，区别在于，
     * thenCombine会将两个任务的执行结果作为方法入参传递到指定方法中，且该方法有返回值；
     * thenAcceptBoth同样将两个任务的执行结果作为方法入参，但是无返回值；
     * runAfterBoth没有入参，也没有返回值。注意两个任务中只要有一个执行异常，则将该异常信息作为指定任务的执行结果。
     *
     * @throws Exception
     */
    public static void testThenCombineAndThenAcceptBothAndRunAfterBoth() throws Exception {
        ForkJoinPool pool=new ForkJoinPool();
        // 创建异步执行任务:
        CompletableFuture<Double> cf = CompletableFuture.supplyAsync(()->{
            System.out.println(Thread.currentThread()+" start job1,time->"+System.currentTimeMillis());
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
            }
            System.out.println(Thread.currentThread()+" exit job1,time->"+System.currentTimeMillis());
            return 1.2;
        });
        CompletableFuture<Double> cf2 = CompletableFuture.supplyAsync(()->{
            System.out.println(Thread.currentThread()+" start job2,time->"+System.currentTimeMillis());
            try {
                Thread.sleep(1500);
            } catch (InterruptedException e) {
            }
            System.out.println(Thread.currentThread()+" exit job2,time->"+System.currentTimeMillis());
            return 3.2;
        });
        //cf和cf2的异步任务都执行完成后，会将其执行结果作为方法入参传递给cf3,且有返回值
        CompletableFuture<Double> cf3=cf.thenCombine(cf2,(a,b)->{
            System.out.println(Thread.currentThread()+" start job3,time->"+System.currentTimeMillis());
            System.out.println("job3 param a->"+a+",b->"+b);
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
            }
            System.out.println(Thread.currentThread()+" exit job3,time->"+System.currentTimeMillis());
            return a+b;
        });

        //cf和cf2的异步任务都执行完成后，会将其执行结果作为方法入参传递给cf3,无返回值
        CompletableFuture cf4=cf.thenAcceptBoth(cf2,(a,b)->{
            System.out.println(Thread.currentThread()+" start job4,time->"+System.currentTimeMillis());
            System.out.println("job4 param a->"+a+",b->"+b);
            try {
                Thread.sleep(1500);
            } catch (InterruptedException e) {
            }
            System.out.println(Thread.currentThread()+" exit job4,time->"+System.currentTimeMillis());
        });

        //cf4和cf3都执行完成后，执行cf5，无入参，无返回值
        CompletableFuture cf5=cf4.runAfterBoth(cf3,()->{
            System.out.println(Thread.currentThread()+" start job5,time->"+System.currentTimeMillis());
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
            }
            System.out.println("cf5 do something");
            System.out.println(Thread.currentThread()+" exit job5,time->"+System.currentTimeMillis());
        });

        System.out.println("main thread start cf.get(),time->"+System.currentTimeMillis());
        //等待子任务执行完成
        System.out.println("cf run result->"+cf.get());
        System.out.println("main thread start cf5.get(),time->"+System.currentTimeMillis());
        System.out.println("cf5 run result->"+cf5.get());
        System.out.println("main thread exit,time->"+System.currentTimeMillis());
    }

    /**
     *
     * applyToEither / acceptEither / runAfterEither
     *
     * 这三个方法都是将两个CompletableFuture组合起来，只要其中一个执行完了就会执行某个任务，其区别在于
     * applyToEither会将已经执行完成的任务的执行结果作为方法入参，并有返回值；
     * acceptEither同样将已经执行完成的任务的执行结果作为方法入参，但是没有返回值；
     * runAfterEither没有方法入参，也没有返回值。注意两个任务中只要有一个执行异常，则将该异常信息作为指定任务的执行结果。
     *
     * @throws Exception
     */
    public static void testApplyToEitherAcceptEitherRunAfterEither() throws Exception {
        // 创建异步执行任务:
        CompletableFuture<Double> cf = CompletableFuture.supplyAsync(()->{
            System.out.println(Thread.currentThread()+" start job1,time->"+System.currentTimeMillis());
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
            }
            System.out.println(Thread.currentThread()+" exit job1,time->"+System.currentTimeMillis());
            return 1.2;
        });
        CompletableFuture<Double> cf2 = CompletableFuture.supplyAsync(()->{
            System.out.println(Thread.currentThread()+" start job2,time->"+System.currentTimeMillis());
            try {
                Thread.sleep(1500);
            } catch (InterruptedException e) {
            }
            System.out.println(Thread.currentThread()+" exit job2,time->"+System.currentTimeMillis());
            return 3.2;
        });
        //cf和cf2的异步任务都执行完成后，会将其执行结果作为方法入参传递给cf3,且有返回值
        CompletableFuture<Double> cf3=cf.applyToEither(cf2,(result)->{
            System.out.println(Thread.currentThread()+" start job3,time->"+System.currentTimeMillis());
            System.out.println("job3 param result->"+result);
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
            }
            System.out.println(Thread.currentThread()+" exit job3,time->"+System.currentTimeMillis());
            return result;
        });

        //cf和cf2的异步任务都执行完成后，会将其执行结果作为方法入参传递给cf3,无返回值
        CompletableFuture cf4=cf.acceptEither(cf2,(result)->{
            System.out.println(Thread.currentThread()+" start job4,time->"+System.currentTimeMillis());
            System.out.println("job4 param result->"+result);
            try {
                Thread.sleep(1500);
            } catch (InterruptedException e) {
            }
            System.out.println(Thread.currentThread()+" exit job4,time->"+System.currentTimeMillis());
        });

        //cf4和cf3都执行完成后，执行cf5，无入参，无返回值
        CompletableFuture cf5=cf4.runAfterEither(cf3,()->{
            System.out.println(Thread.currentThread()+" start job5,time->"+System.currentTimeMillis());
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
            }
            System.out.println("cf5 do something");
            System.out.println(Thread.currentThread()+" exit job5,time->"+System.currentTimeMillis());
        });

        System.out.println("main thread start cf.get(),time->"+System.currentTimeMillis());
        //等待子任务执行完成
        System.out.println("cf run result->"+cf.get());
        System.out.println("main thread start cf5.get(),time->"+System.currentTimeMillis());
        System.out.println("cf5 run result->"+cf5.get());
        System.out.println("main thread exit,time->"+System.currentTimeMillis());
    }

    /**
     * thenCompose
     * thenCompose方法会在某个任务执行完成后，将该任务的执行结果作为方法入参然后执行指定的方法，
     * 该方法会返回一个新的CompletableFuture实例，如果该CompletableFuture实例的result不为null，
     * 则返回一个基于该result的新的CompletableFuture实例；如果该CompletableFuture实例为null，
     * 则，然后执行这个新任务
     *
     * @throws Exception
     */
    public static void testThenCompose() throws Exception {

        // 创建异步执行任务:
        CompletableFuture<Double> cf = CompletableFuture.supplyAsync(()->{
            System.out.println(Thread.currentThread()+" start job1,time->"+System.currentTimeMillis());
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
            }
            System.out.println(Thread.currentThread()+" exit job1,time->"+System.currentTimeMillis());
            return 1.2;
        });
        CompletableFuture<String> cf2= cf.thenCompose((param)->{
            System.out.println(Thread.currentThread()+" start job2,time->"+System.currentTimeMillis());
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
            }
            System.out.println(Thread.currentThread()+" exit job2,time->"+System.currentTimeMillis());
            return CompletableFuture.supplyAsync(()->{
                System.out.println(Thread.currentThread()+" start job3,time->"+System.currentTimeMillis());
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                }
                System.out.println(Thread.currentThread()+" exit job3,time->"+System.currentTimeMillis());
                return "job3 test";
            });
        });
        System.out.println("main thread start cf.get(),time->"+System.currentTimeMillis());
        //等待子任务执行完成
        System.out.println("cf run result->"+cf.get());
        System.out.println("main thread start cf2.get(),time->"+System.currentTimeMillis());
        System.out.println("cf2 run result->"+cf2.get());
        System.out.println("main thread exit,time->"+System.currentTimeMillis());
    }

    /**
     *
     * allOf / anyOf
     * allOf返回的CompletableFuture是多个任务都执行完成后才会执行，只有有一个任务执行异常，
     * 则返回的CompletableFuture执行get方法时会抛出异常，如果都是正常执行，则get返回null。
     *
     * @throws Exception
     */
    public static void testAllOf() throws Exception {
            // 创建异步执行任务:
            CompletableFuture<Double> cf = CompletableFuture.supplyAsync(()->{
                System.out.println(Thread.currentThread()+" start job1,time->"+System.currentTimeMillis());
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                }
                int i = 10 / 0;
                System.out.println(Thread.currentThread()+" exit job1,time->"+System.currentTimeMillis());
                return 1.2;
            });
            CompletableFuture<Double> cf2 = CompletableFuture.supplyAsync(()->{
                System.out.println(Thread.currentThread()+" start job2,time->"+System.currentTimeMillis());
                try {
                    Thread.sleep(1500);
                } catch (InterruptedException e) {
                }
                System.out.println(Thread.currentThread()+" exit job2,time->"+System.currentTimeMillis());
                return 3.2;
            });
            CompletableFuture<Double> cf3 = CompletableFuture.supplyAsync(()->{
                System.out.println(Thread.currentThread()+" start job3,time->"+System.currentTimeMillis());
                try {
                    Thread.sleep(1300);
                } catch (InterruptedException e) {
                }
//            throw new RuntimeException("test");
                System.out.println(Thread.currentThread()+" exit job3,time->"+System.currentTimeMillis());
                return 2.2;
            });
            //allof等待所有任务执行完成才执行cf4，如果有一个任务异常终止，则cf4.get时会抛出异常，都是正常执行，cf4.get返回null
            //anyOf是只有一个任务执行完成，无论是正常执行或者执行异常，都会执行cf4，cf4.get的结果就是已执行完成的任务的执行结果
            CompletableFuture cf4=CompletableFuture.allOf(cf,cf2,cf3).whenComplete((a,b)->{
            if(b!=null){
                System.out.println("error stack trace->");
                b.printStackTrace();
            }else{
                System.out.println("run succ,result->"+a);
            }
        });

            System.out.println("main thread start cf4.get(),time->"+System.currentTimeMillis());
            //等待子任务执行完成
            System.out.println("cf4 run result->"+cf4.get());
            Thread.sleep(1000);
            System.out.println("main thread exit,time->"+System.currentTimeMillis());

    }
}
